package juc;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 使用 生产者 -> 队列 -> 消费者 ,测试多 condition 场景.
 */
public class TestMultipleConditions {
    private static final Queue<Integer> QUEUE = new LinkedList<>();
    private static final Integer MAX_SIZE = 10;
    private static final Lock produceLock = new ReentrantLock();
    private static final Lock consumerLock = new ReentrantLock();
    private static final Condition notFull = produceLock.newCondition();
    private static final Condition notEmpty = consumerLock.newCondition();

    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 2,
                0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(1));
//        ExecutorService threadPoolExecutor = Executors.newFixedThreadPool(2);
        Producer producer = new Producer();
        Consumer consumer = new Consumer();
        boolean flag = true;
        threadPoolExecutor.submit(() -> {
            for (int i = 0; i < 100; i++) {
                producer.produce(i);
                try {
                    Thread.sleep(100L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        threadPoolExecutor.submit(() -> {
            for (; ; ) {
                consumer.consume();
                try {
                    Thread.sleep(200L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    static class Producer {
        public void produce(Integer record) {
            produceLock.lock();
            int currentCount = -1;
            try {
                if (QUEUE.size() == MAX_SIZE) {
                    //  如果队列满了等待消费者消费
                    produceLock.wait();
                    notFull.await();
                }
                currentCount = QUEUE.size();
                QUEUE.offer(record);
                System.out.println("producer: " + record);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                produceLock.unlock();
            }
            if (currentCount == 0) {
                notifyConsumer();
            }
        }
    }

    static class Consumer {
        public void consume() {
            consumerLock.lock();
            try {
                if (QUEUE.size() == 0) {
                    notEmpty.await();
                }
                Integer poll = QUEUE.poll();
                System.out.println("consumer: " + poll);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                consumerLock.unlock();
            }
            notifyProducer();
        }
    }

    private static void notifyProducer() {
        produceLock.lock();
        try {
            notFull.signal();
        } finally {
            produceLock.unlock();
        }
    }

    private static void notifyConsumer() {
        consumerLock.lock();
        try {
            notEmpty.signal();
        } finally {
            consumerLock.unlock();
        }
    }
}
